package com.bw.data_summary;

import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.TableResult;

//  门禁类型
public class Dwd_entrance_access_control {
    public static void main(String[] args) throws Exception {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

//        3.读取门禁刷卡事实表ods_entrance_guard_log
       tEnv.executeSql("CREATE TABLE `bean_entrance_guard_log`(\n" +
                " id STRING,\n" +
                " openMode INT,\n" +
                " openTime STRING,\n" +
                " ownerId INT,\n" +
                " proc_time as PROCTIME()\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'entrance_guard_log',\n" +
                " 'properties.bootstrap.servers' = 'node102:9092',\n" +
                " 'properties.group.id' = 'entrance_guard_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");


//        tEnv.executeSql("SELECT * FROM bean_entrance_guard_log").print();

        //4.业主维表
      tEnv.executeSql("CREATE TABLE dim_base_ower(\n" +
                " id INT,\n" +
                " roomId INT,\n" +
                " owerCode STRING,\n" +
                " owerName STRING,\n" +
                " owerSex STRING,\n" +
                " owerAge INT,\n" +
                " owerIdcard STRING,\n" +
                " owerTel STRING,\n" +
                " owerHomeplace STRING,\n" +
                " owerWorkplace STRING,\n" +
                " remark STRING,\n" +
                " createTime STRING,\n" +
                " operatorTime STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
              " 'connector' = 'doris',\n" +
              " 'fenodes' = 'node102:8030',\n" +
              " 'jdbc-url' = 'jdbc:mysql://node102:9092?useSSL=false',  \n" +
              " 'username' = 'root',\n" +
              " 'password' = '123456',\n" +
              " 'table.identifier' = 'sca_doris.ower',\n" +
              " 'lookup.cache.max-rows' = '10000',\n" +
              " 'lookup.cache.ttl' = '60s'\n" +
                ")\n");

//        tEnv.executeSql("SELECT * FROM dim_base_ower").print();

        //5. 房间维度表
        tEnv.executeSql("CREATE TABLE dim_base_room(\n" +
                " id INT,\n" +
                " houseId INT,\n" +
                " roomCode STRING,\n" +
                " roomName STRING,\n" +
                " peopleNum INT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                " 'fenodes' = 'node102:8030',\n" +
                " 'jdbc-url' = 'jdbc:mysql://node102:9092?useSSL=false',  \n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'table.identifier' = 'sca_doris.room',\n" +
                " 'lookup.cache.max-rows' = '10000',\n" +
                " 'lookup.cache.ttl' = '60s'\n" +
                ")\n");
//        tEnv.executeSql("SELECT * FROM dim_base_room").print();


        //6. 楼栋维度表
        tEnv.executeSql("CREATE TABLE dim_base_house(\n" +
                " id INT,\n" +
                " cid INT,\n" +
                " houseCode STRING,\n" +
                " houseName STRING,\n" +
                " buildStartTime STRING,\n" +
                " buildEndTime STRING,\n" +
                " houseArea FLOAT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'doris',\n" +
                " 'fenodes' = 'node102:8030',\n" +
                " 'jdbc-url' = 'jdbc:mysql://node102:9092?useSSL=false',  \n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'table.identifier' = 'sca_doris.house',\n" +
                " 'lookup.cache.max-rows' = '10000',\n" +
                " 'lookup.cache.ttl' = '60s'\n" +
                ")\n");
//        tEnv.executeSql("SELECT * FROM dim_base_house").print();

        //7. 小区维度表
        tEnv.executeSql("CREATE TABLE dim_base_community(\n" +
                " id INT,\n" +
                " communityName STRING,\n" +
                " district_id INT,\n" +
                " city_id INT,\n" +
                " communityLocation STRING,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                " 'fenodes' = 'node102:8030',\n" +
                " 'jdbc-url' = 'jdbc:mysql://node102:9092?useSSL=false',  \n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'table.identifier' = 'sca_doris.community',\n" +
                " 'lookup.cache.max-rows' = '10000',\n" +
                " 'lookup.cache.ttl' = '60s'\n" +
                ")\n");

//        tEnv.executeSql("SELECT * FROM dim_base_community").print();

        // 地区维度表
        tEnv.executeSql("CREATE TABLE dim_base_district (\n" +
                " id INT,\n" +
                " pid INT,\n" +
                " code STRING,\n" +
                " name STRING,\n" +
                " remark STRING,\n" +
                " create_time TIMESTAMP,\n" +
                " update_time TIMESTAMP,\n" +
                " status TINYINT,\n" +
                " level TINYINT,\n" +
                " PRIMARY KEY (`id`)  NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                " 'fenodes' = 'node102:8030',\n" +
                " 'jdbc-url' = 'jdbc:mysql://node102:9092?useSSL=false',  \n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'table.identifier' = 'sca_doris.district',\n" +
                " 'lookup.cache.max-rows' = '10000',\n" +
                " 'lookup.cache.ttl' = '60s'\n" +
                ")\n");

//        tEnv.executeSql("SELECT * FROm dim_base_district").print();

        //8.事实表与维度表关联

        TableResult resultTable = tEnv.executeSql("select " +
                "b1.id, " +
                "b1.ownerId, " +
                "b1.openTime, " +
                "d3.cid as cId, " +
                "d3.houseCode, " +
                "d2.roomCode, " +
                "b1.openMode " +
                "from bean_entrance_guard_log b1 " +
                "left join dim_base_ower for system_time as of b1.proc_time as d1 " +
                "on b1.ownerId=d1.id and b1.ownerId is not null " +
                "left join dim_base_room for system_time as of b1.proc_time as d2 " +
                "on d1.roomId=d2.id and d1.roomId is not null " +
                "left join dim_base_house for system_time as of b1.proc_time as d3 " +
                "on d2.houseId=d3.id and d2.houseId is not null " +
                "left join dim_base_community for system_time as of b1.proc_time as d4 " +
                "on d3.cid=d4.id and d3.cid is not null "
        );

        resultTable.print();

//        tEnv.createTemporaryView("resultTable", resultTable);


        //9.创建dwd_entrance_guard_log表
        tEnv.executeSql("CREATE TABLE dwd_entrance_guard_log (\n" +
                "id STRING,\n" +
                "ownerId INT,\n" +
                "openTime STRING,\n" +
                "cId INT,\n" +
                "houseCode STRING,\n" +
                "roomCode STRING,\n" +
                "openMode INT,\n" +
                "districtName STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'upsert-kafka',\n" +
                " 'topic' = 'dwd_entrance_guard_log2',\n" +
                " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                " 'key.format' = 'json',\n" +
                " 'value.format' = 'json'\n" +
                ")");

        //10.flink sql 将查询结果写入dwd_entrance_guard_log
        tEnv.executeSql("INSERT INTO dwd_entrance_guard_log SELECT * FROM resultTable");


        // 提交Flink作业
//        senv.execute("dwd_entrance_guard_log");
    }

}
